/*
 * Created by SharpDevelop.
 * User: Administrator
 * Date: 6/3/2009
 * Time: 12:12 PM
 * 
 * To change this template use Tools | Options | Coding | Edit Standard Headers.
 */
using System;
using System.Collections;
using System.Collections.Generic;
using System.IO;
using System.Reflection;
using System.Runtime.Serialization;

using Communication;
using Communication.Network;
using System.Threading;

namespace MapSharpLib
{	
	public class MRNetworkNode
	{
		private Server<Job> ss;
		private MRNActor m;
		
		public MRNetworkNode(int port)
		{
			m = new MRNActor(new asyncTransfer<Job>(), null);
			ss = new Server<Job>(port,m);
		}
		
		public void Stop()
		{
			m.Stop();
			ss.Stop();
		}
		
	}
	
	
	public class MRNode
	{
		public static Job Run(Job j)
		{
			Type MapReducerType = DllLoader.LoadDll(j.Assembly,j.MapReducerClass);
			MapperReducer mr = (MapperReducer)Activator.CreateInstance(MapReducerType);
			IEnumerable<ISerializable> inputs = j.Inputs;
			
			ISerializable output = MapReducer.MapReduce(mr, inputs);
			
			j.SetResults(output);
			return j;
		}
		
		public static Job Reduce(IList<Job> jA)
		{
			Job j = jA[0];
			Type MapReducerType = DllLoader.LoadDll(j.Assembly,j.MapReducerClass);
			MapperReducer mr = (MapperReducer)Activator.CreateInstance(MapReducerType);
			
			List<ISerializable> isL = new List<ISerializable>(jA.Count);
			
			foreach(Job ij in jA)
				isL.Add(ij.Results);
			
			ISerializable res = MapReducer.Reduce(mr,isL);
			Job retVal = j.Clone();
			retVal.SetResults(res);
			return 	retVal;
		}
	}
	
	public class MRNActor : IActor<Job>
	{
		ObjectPipe<Job> pipe;
		volatile Queue<Job> wq;
		Thread[] threads;
		
		
		public MRNActor(ObjectPipe<Job> dp, Queue<Job> WorkQueue)
		{
			if(WorkQueue ==null)
			{
				//Basically, This is a new Node. So make a new work queue
				//and spawn a pair of work threads.
				this.wq = new Queue<Job>();
				
				//HACK: Hard-coded variable: 2 threads per Node
				int numthreads = 2;
				
				threads = new Thread[numthreads];
				for(int i=0;i<numthreads;i++)
				{
					threads[i] = new Thread(new ThreadStart(this.worker));
					threads[i].Start();
				}
			}
			else
				this.wq = WorkQueue;
			
			this.pipe = dp;
		}
		
		public void Stop()
		{
			foreach(Thread t in threads)
				t.Abort();
		}
				
		private void worker()
		{
			while(true)
			{
				if(wq!=null && wq.Count>0)
				{
					Job workjob = null;
					lock(wq)
					{
						if(wq.Count>0)
							workjob = wq.Dequeue();
					}
					if(workjob!=null)
					{
						Job j = MRNode.Run(workjob);
						pipe.pushObject(j.Client,j);
					}
				}
				Thread.Sleep(500);
			}
		}
		
		public IActor<Job> newActor(ObjectPipe<Job> dp)
		{
			return new MRNActor(dp, wq);
		}
		
		public void Act()
		{
			Job j = pipe.getObject();
			lock(wq)
			{
				wq.Enqueue(j);
			}
		}
	}
	
	public class DllLoader
	{
		public static Type LoadDll(string DLLpath, string MRclass)
		{
			try
			{
				FileInfo fi = new FileInfo(DLLpath);
				string fullDLLpath = fi.FullName;
				Assembly a = Assembly.LoadFile(fullDLLpath);
				Type t = a.GetType(MRclass);
				return t;
			}
			catch(Exception e)
			{
				Exception ex = new Exception("method/class not found",e);
				throw ex;
			}
		}
		
		public static Type LoadDll(byte[] rawAssembly, string MRclass)
		{
			try
			{
				Assembly a = Assembly.Load(rawAssembly);
				Type t = a.GetType(MRclass);
				return t;
			}
			catch(Exception e)
			{
				Exception ex = new Exception("method/class not found",e);
				throw ex;
			}
		}
	}
}